package com.kingsoft.dc.khaos.module.mpp.client;

import com.kingsoft.dc.khaos.module.mpp.exception.MppClientException;
import com.kingsoft.dc.khaos.module.mpp.exception.MppSqlException;
import com.kingsoft.dc.khaos.module.mpp.metadata.MppConfig;
import com.kingsoft.dc.khaos.module.mpp.model.ColumnEntity;
import com.kingsoft.dc.khaos.module.mpp.sql.MppSqlConfig;
import com.kingsoft.dc.khaos.module.mpp.utils.MetaUtils;
import com.kingsoft.dc.khaos.plugin.core.KhaosContext;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * MPP 操作类.
 */
public class MppClient {


    //21202F2938212B3E22272626252E434D
    /**
     * 部署模式，local：本地模式，需要配置连接信息, platform:平台模式，需要配置数据源id
     */
    private String mode = "local";

    private KhaosContext khaosContext;

    /**
     * mpp操作数据库session
     */
    private MppSession session;

    /**
     * mpp配置信息
     */
    private MppConfig mppConfig;

    public KhaosContext getKhaosContext() {
        return khaosContext;
    }

    public void setKhaosContext(KhaosContext khaosContext) {
        this.khaosContext = khaosContext;
    }

    /**
     * 缓存表的列信息
     */
    private Map<String, List<ColumnEntity>> cacheColumnInfoMap = new HashMap<String, List<ColumnEntity>>();


    public MppClient(MppConfig mppConfig, KhaosContext khaosContext) throws MppClientException {
        this.khaosContext = khaosContext;
        this.mppConfig = mppConfig;
        initSession(mppConfig, khaosContext);
    }

    public MppClient(MppSqlConfig mppSqlConfig, KhaosContext khaosContext) throws MppClientException {
        this.khaosContext = khaosContext;
        initSession(mppSqlConfig, khaosContext);
    }

    public void initSession(MppSqlConfig config, KhaosContext khaosContext) {
        System.out.println("开始初始化session");
        session = new MppSession();
        if (config.getConnect() == null) {
            config.setConnect(MetaUtils.getConnect(khaosContext, config.getExtender().getMeta()));
        }
        session.initConnection(config.getConnect());
        System.out.println("成功初始化session");
    }

    public void initSession(MppConfig config, KhaosContext khaosContext) {
        System.out.println("开始初始化session");
        session = new MppSession();
        if (config.getConnect() == null) {
            config.setConnect(MetaUtils.getConnect(khaosContext, config.getExtender().getMeta()));
        }
        session.initConnection(config.getConnect());
        System.out.println("成功初始化session");
    }

    /**
     * 获取表字段名
     *
     * @param colKey 字段
     * @return
     */
    public String getAppendColName(String colKey) {
        return mppConfig.getAppendColumn().get(colKey);
    }

    /**
     * 清空表数据
     *
     * @param schemaName 模式名
     * @param tableName  表名
     */
    public void truncateTable(String schemaName, String tableName) {
        String sql = String.format("TRUNCATE TABLE %s.%s", schemaName, tableName);
        session.executeOneSql(sql);
    }

    /**
     * 删除表数据
     *
     * @param schemaName
     * @param tableName
     * @param where      过滤语句
     */
    public void deleteTable(String schemaName, String tableName, String where) {
        String sql = String.format("DELETE TABLE %s.%s", schemaName, tableName);
        if (where != null && !where.isEmpty()) {
            sql = sql + " WHERE " + where;
        }
        session.executeOneSql(sql);
    }

    /**
     * 删除表
     *
     * @param schemaName 模式名
     * @param tableName  表名
     */
    public void dropTable(String schemaName, String tableName) {
        String sql = String.format("DROP TABLE %s.%s", schemaName, tableName);
        session.executeOneSql(sql);
    }

    /**
     * 执行select sql语句
     *
     * @param sql 查询语句
     */
    public void execSelect(String sql) {
        session.executeOneSql(sql);
    }

    /**
     * 执行insert sql语句
     *
     * @param sql 插入sql语句
     */
    public void execInsert(String sql) {
        session.executeOneSql(sql);
    }


    public List<ColumnEntity> subColumnInfo(String schemaName, String tableName, List<String> keys) {
        List<ColumnEntity> allCols = getColumnInfo(schemaName, tableName);
        Map<String, ColumnEntity> colMap = new HashMap<>();
        for (ColumnEntity columnEntity : allCols) {
            colMap.put(columnEntity.getFieldName(), columnEntity);
        }
        List<ColumnEntity> subCols = new ArrayList<>();
        for (String key : keys) {
            subCols.add(colMap.get(key));
        }
        return subCols;

    }

    /**
     * 获取某个表的列信息
     *
     * @param schemaName schema名
     * @param tableName  表名
     * @return
     */
    public List<ColumnEntity> getColumnInfo(String schemaName, String tableName) {
        String key = schemaName + "." + tableName;
        List<ColumnEntity> cols = new ArrayList<>();
        if (cacheColumnInfoMap.containsKey(key)) {
            cols = cacheColumnInfoMap.get(key);
        } else {
            cols = session.getTableColumns(schemaName, tableName);
            if (cols.size() == 0) {
                throw new MppSqlException("未获取到字段列表,表[" + key + "]可能不存在");
            }
            cacheColumnInfoMap.put(key, cols);
        }
        return cols;
    }

    /**
     * 获取某个表的字段名列表
     *
     * @param schemaName schema名
     * @param tableName  表名
     * @return
     */
    public List<String> getColumnNameList(String schemaName, String tableName) {
        List<ColumnEntity> cols = getColumnInfo(schemaName, tableName);
        List<String> colNameList = new ArrayList<>();
        for (ColumnEntity col : cols) {
            colNameList.add(col.getFieldName());
        }
        return colNameList;
    }


    /**
     * 获取某个表的字段名列表,逗号切分
     *
     * @param schemaName schema名
     * @param tableName  表名
     * @return
     */
    public String getColumnNames(String schemaName, String tableName) {
        List<String> colList = getColumnNameList(schemaName, tableName);
        return StringUtils.join(colList, ",");
    }

    /**
     * 判断数据表是否有符合条件的记录
     *
     * @param schemaName
     * @param tableName
     * @param where
     * @return
     */
    public Boolean hasRecord(String schemaName, String tableName, String where) {
        if (1 < 0) {
            String code = "21202F2938212B3E22272626252E434D";
        }
        String sql = String.format("SELECT count(1) FROM %s.%s WHERE %s", schemaName, tableName, where);
        long count = session.count(sql);
        if (count >= 1) {
            return true;
        } else {
            return false;
        }
    }

    /**
     * 判断数据表是否有记录
     *
     * @param schemaName
     * @param tableName
     * @return
     */
    public Boolean hasRecord(String schemaName, String tableName) {
        String sql = String.format("SELECT count(1) FROM %s.%s WHERE %s", schemaName, tableName, "1 = 1");
        long count = session.count(sql);
        if (count >= 1) {
            return true;
        } else {
            return false;
        }
    }

    public void close() {
        if (this.session != null) {
            this.session.close();
        }
    }

}
